package com.lagou.rpc.consumer;

import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import com.lagou.rpc.registry.RegistryDiscoveryCenter;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.RetryNTimes;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * 客户端测试类
 */
public class ClientBootStrap {
    // 所有的服务器列表
    private static List<String> serviceList = new ArrayList<>();
    private static CuratorFramework client;
    public static void main(String[] args) throws Exception {
        // 读取zk节点信息，初始化目标服务器列表
        initServers();
        // 与每个服务器建立连接生成RpcClient备用
        initRpcClients();
        test();
    }



    // 找到/rpc-providers的所有子节点，每个子节点对应一个rpc-provider
    private static void initServers() throws Exception {
        if (!serviceList.isEmpty()) {
            return;
        }

        RetryPolicy retryPolicy = new RetryNTimes(1, 1000);
        client = CuratorFrameworkFactory.builder()
                .connectString("localhost:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .build();

        client.start();

        System.out.println(String.format("客户端连接至zk服务器成功"));

        // 读取zk根节点/rpc-providers下的每一个子节点对应一个rpc-provider
        String rootNodePath = RegistryDiscoveryCenter.ZK_RPC_SERVERS_ROOT_PATH;
        if (serviceList.isEmpty()) {
            serviceList = client.getChildren().forPath(rootNodePath);
        }
        System.out.println("服务器列表信息：" + serviceList);
        // 注册监听
        registerWatcher(rootNodePath);
    }


    // 监听providers的变化
    private static void registerWatcher(final String path) {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        try {
            /*
             * StartMode：初始化方式
             * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
             * NORMAL：异步初始化
             * BUILD_INITIAL_CACHE：同步初始化
             * */
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            pathChildrenCache.getListenable().addListener((client, pathChildrenCacheEvent) -> {
                // 当providers列表变化时，这里的serviceList要重新获取实时更新
                serviceList = client.getChildren().forPath(path);
                System.out.println("providers有变化，变化后:" + serviceList);
                handleChildrenNodesChangeEvent(pathChildrenCacheEvent);
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void handleChildrenNodesChangeEvent(PathChildrenCacheEvent pathChildrenCacheEvent) {
        PathChildrenCacheEvent.Type eventType = pathChildrenCacheEvent.getType();
        System.out.println("收到节点变更通知:" + eventType);
        String path = pathChildrenCacheEvent.getData().getPath();
        System.out.println("变化节点path=" + path);
        String serverInfo = path.substring(path.lastIndexOf("/") + 1);
        String[] spit = serverInfo.split(":");
        String ip = spit[0];
        int port = Integer.parseInt(spit[1]);
        // 增加节点，服务上线
        if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(eventType)) {
            // 如果检测服务上线，要新建一个RpcClient以备后续使用
            RpcClientProxy.addRpcClient(ip, port);
            System.out.println("新增节点:" + serverInfo);
        } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(eventType)) {
            // 如果检测服务下线，要移除对应的RpcClient
            RpcClientProxy.removeRpcClient(serverInfo);
            System.out.println("移除节点:" + serverInfo);
        }
    }

    private static void initRpcClients() {
        for (String connectStr : serviceList) {
            String[] split = connectStr.split(":");
            String ip = split[0];
            int port = Integer.parseInt(split[1]);
            RpcClientProxy.addRpcClient(ip, port);
        }
    }

    private static void test() throws InterruptedException {
        while (true) {
            TimeUnit.SECONDS.sleep(3);
            // 找到目标服务器
            Optional<String> optional = findTargetServer();
            optional.ifPresent(targetServer -> {
                System.out.println("目标服务器为:" + targetServer);
                String ip = targetServer.split(":")[0];
                int port = Integer.parseInt(targetServer.split(":")[1]);
                IUserService userService = (IUserService) RpcClientProxy.createProxy(IUserService.class, ip, port);
                userService.getById(1);
            });
            System.out.println("==================================");
        }
    }

    private static Optional<String> findTargetServer() {
        if (serviceList.isEmpty()) {
            System.out.println("暂无可用的服务器");
            return Optional.empty();
        }
        // 找到耗时最少的服务器
        String target = serviceList.get(0);
        long minCostTime = Long.MAX_VALUE;
        for (String server : serviceList) {
            long costTime = ServerRequestCostTimeCache.getLastCostTime(server);
            System.out.println(server + "上次响应时间为:" + costTime + "ms");
            if (costTime < minCostTime) {
                minCostTime = costTime;
                target = server;
            }
        }
        return Optional.of(target);
    }
}
